package cn.zhangtiancinb.simple_demo.http.utils;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

import javax.annotation.PostConstruct;

import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.beans.factory.annotation.Value;

/**
 * 消息消费工具类
 *
 * @author 张天赐
 */
@Slf4j
@Component
public class MessageConsumeUtil {
    /**
     * 消费线程池
     */
    private ExecutorService pool;
    /**
     * 消费延时锁
     */
    private CyclicBarrier barrier;
    /**
     * 消费延时条数
     */
    @Value("${mq.delayed.consume_once_number}")
    int consumerOnceNumber;

    /**
     * 类初始化
     */
    @PostConstruct
    void init() {
        barrier = new CyclicBarrier(consumerOnceNumber, () -> log.warn("consume barrier break through once."));
        AtomicInteger threadPoolIndex = new AtomicInteger();
        pool = new ThreadPoolExecutor(consumerOnceNumber, consumerOnceNumber,
            0L, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<>(),
            r -> new Thread(r) {{setName("ConsumePoolThread:" + threadPoolIndex.getAndAdd(1));}});
    }

    /**
     * 提交消息
     *
     * @param message 消息内容
     */
    public void submit(String message) {
        pool.submit(new ConsumeRunnable(message));
    }

    /**
     * 自定义-处理消息-线程可运行对象
     *
     * @author 张天赐
     */
    private class ConsumeRunnable implements Runnable {

        /**
         * 消息内容
         */
        private String message;

        /**
         * 隐藏无参构造函数
         */
        private ConsumeRunnable() {}

        /**
         * 带参构造函数
         *
         * @param message 消息内容
         */
        ConsumeRunnable(String message) {
            this.message = message;
        }

        /**
         * 可以被调用的运行方法
         */
        @Override
        public void run() {
            try {
                barrier.await();
                log.info("consume:{}", message);
            } catch (Exception e) {
                log.error("consume happen exception", e);
            }
        }
    }

}
